package he.config;

import io.debezium.engine.ChangeEvent;
import io.debezium.engine.DebeziumEngine;
import io.debezium.engine.format.Json;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.KafkaTemplate;

import java.util.Properties;

@Configuration
public class DebeziumConfig {

    @Bean
    public DebeziumEngine<ChangeEvent<String, String>> debeziumEngine(KafkaTemplate<String, String> kafkaTemplate) {
        Properties props = new Properties();
        props.setProperty("connector.class", "io.debezium.connector.mysql.MySqlConnector");
        props.setProperty("database.hostname", "localhost");
        props.setProperty("database.port", "3306");
        props.setProperty("database.user", "root");
        props.setProperty("database.password", "123456");
        props.setProperty("database.allowPublicKeyRetrieval", "true");

        // 添加SSL配置（关键修复1）
        props.setProperty("database.ssl.mode", "PREFERRED");

        props.setProperty("database.server.name", "shop-mysql-server");
        props.setProperty("database.server.id", "184064");
        props.setProperty("database.include.list", "shopdb");
        props.setProperty("database.table.include.list", "shopdb.product");
        props.setProperty("database.server.timeZone", "UTC");
        props.setProperty("database.time.precision.mode", "connect");

        // 修改快照模式（关键修复2）
        props.setProperty("snapshot.mode", "when_needed");

        // 添加权限配置（关键修复3）
        props.setProperty("database.replication.user", "debezium");
        props.setProperty("database.replication.password", "dbz_password");

        // 移除错误的连接池配置（关键修复4）
        // props.setProperty("database.connection.adapter", "...");
        // props.setProperty("database.connection.adapter.connection.string", "...");

        // 添加连接超时配置（关键修复5）
        props.setProperty("connect.timeout.ms", "30000");
        props.setProperty("connection.adapter.max.pool.size", "5");

        // 添加binlog配置（关键修复6）
        props.setProperty("database.server.id", "184064");
        props.setProperty("database.server.name", "shop-mysql-server");
        props.setProperty("database.include.list", "shopdb");
        props.setProperty("database.binlog", "mysql-bin");
        props.setProperty("database.binlog.index", "mysql-bin.index");

        // 指定使用 JSON 格式
        props.setProperty("key.converter", "org.apache.kafka.connect.storage.JsonConverter");
        props.setProperty("value.converter", "org.apache.kafka.connect.storage.JsonConverter");

        return DebeziumEngine.create(Json.class)
                .using(props)
                .notifying(record -> {
                    // record 是 Json 类型
                    String topic = record.destination();
                    String key = record.key().toString();
                    String value = record.value().toString();

                    kafkaTemplate.send(topic, key, value);
                })
                .build();
    }
}
